#include <gtest/gtest.h>
#include <gflags/gflags.h>
#include <string>
#include <set>

#include <boost/date_time/posix_time/posix_time.hpp>

#include "mysql_outbox.hpp"
#include "utils.hpp"

DEFINE_string(outbox_mysql_user, "root", "Outbox 测试 MySQL 用户名");
DEFINE_string(outbox_mysql_password, "ok", "Outbox 测试 MySQL 密码");
DEFINE_string(outbox_mysql_db, "suye", "Outbox 测试 MySQL 数据库名称");
DEFINE_string(outbox_mysql_host, "127.0.0.1", "Outbox 测试 MySQL 主机地址");
DEFINE_string(outbox_mysql_charset, "utf8", "Outbox 测试 MySQL 字符集");
DEFINE_int32(outbox_mysql_port, 0, "Outbox 测试 MySQL 端口，为 0 时使用默认端口");
DEFINE_int32(outbox_mysql_pool_size, 1, "Outbox 测试连接池大小");

namespace
{
class OutboxTableTest : public ::testing::Test
{
protected:
    static void SetUpTestSuite()
    {
        _db = suye::ODBFactory::create(FLAGS_outbox_mysql_user,
                                       FLAGS_outbox_mysql_password,
                                       FLAGS_outbox_mysql_db,
                                       FLAGS_outbox_mysql_host,
                                       FLAGS_outbox_mysql_charset,
                                       FLAGS_outbox_mysql_port,
                                       FLAGS_outbox_mysql_pool_size);
        ASSERT_TRUE(_db);
        _table = std::make_shared<suye::OutboxTable>(_db);
        ASSERT_TRUE(_table->truncate());
    }

    static void TearDownTestSuite()
    {
        if (_table)
        {
            _table->truncate();
        }
        _table.reset();
        _db.reset();
    }

    void SetUp() override
    {
        ASSERT_TRUE(_table->truncate());
    }

    static std::shared_ptr<odb::core::database> _db;
    static std::shared_ptr<suye::OutboxTable> _table;
};

std::shared_ptr<odb::core::database> OutboxTableTest::_db;
std::shared_ptr<suye::OutboxTable> OutboxTableTest::_table;
} // namespace

TEST_F(OutboxTableTest, RecordEventCommit)
{
    const std::string event_id = suye::uuid();
    auto event = std::make_shared<suye::OutboxEvent>(
        event_id,
        "user",
        "uid-123",
        "profile_updated",
        "{\"field\":\"nickname\",\"new\":\"test\"}");

    {
        odb::transaction trans(OutboxTableTest::_db->begin());
        ASSERT_TRUE(OutboxTableTest::_table->record_event(event));
        trans.commit();
    }

    auto stored = OutboxTableTest::_table->find_by_event_id(event_id);
    ASSERT_TRUE(stored);
    EXPECT_EQ(stored->event_id(), event_id);
    EXPECT_EQ(stored->aggregate_type(), "user");
    EXPECT_EQ(stored->aggregate_id(), "uid-123");
    EXPECT_EQ(stored->event_type(), "profile_updated");
    EXPECT_EQ(stored->status(), suye::OutboxStatus::Pending);
}

TEST_F(OutboxTableTest, RecordEventRollback)
{
    const std::string event_id = suye::uuid();
    auto event = std::make_shared<suye::OutboxEvent>(
        event_id,
        "user",
        "uid-rollback",
        "profile_updated",
        "{\"field\":\"desc\",\"new\":\"rollback\"}");

    {
        odb::transaction trans(OutboxTableTest::_db->begin());
        ASSERT_TRUE(OutboxTableTest::_table->record_event(event));
        // 未显式提交，析构时自动回滚
    }

    auto stored = OutboxTableTest::_table->find_by_event_id(event_id);
    EXPECT_FALSE(stored);
}

TEST_F(OutboxTableTest, ClaimBatchAndMarkProcessed)
{
    std::vector<std::string> event_ids;
    for (int i = 0; i < 3; ++i)
    {
        const std::string event_id = suye::uuid();
        event_ids.emplace_back(event_id);
        auto event = std::make_shared<suye::OutboxEvent>(
            event_id,
            "user",
            "uid-" + std::to_string(i),
            "profile_updated",
            "{\"field\":\"nickname\",\"new\":\"claim\"}");

        odb::transaction trans(OutboxTableTest::_db->begin());
        ASSERT_TRUE(OutboxTableTest::_table->record_event(event));
        trans.commit();
    }

    auto batch = OutboxTableTest::_table->claim_batch(2);
    ASSERT_EQ(batch.size(), 2u);

    std::set<std::string> claimed_ids;
    for (const auto& event_ptr : batch)
    {
        claimed_ids.insert(event_ptr->event_id());
        auto stored = OutboxTableTest::_table->find_by_event_id(event_ptr->event_id());
        ASSERT_TRUE(stored);
        EXPECT_EQ(stored->status(), suye::OutboxStatus::Dispatching);
    }

    for (const auto& event_id : claimed_ids)
    {
        EXPECT_TRUE(OutboxTableTest::_table->mark_processed(event_id));
        auto stored = OutboxTableTest::_table->find_by_event_id(event_id);
        ASSERT_TRUE(stored);
        EXPECT_EQ(stored->status(), suye::OutboxStatus::Completed);
    }

    auto remaining_batch = OutboxTableTest::_table->claim_batch(5);
    ASSERT_EQ(remaining_batch.size(), 1u);
    const std::string remaining_id = remaining_batch.front()->event_id();

    EXPECT_TRUE(OutboxTableTest::_table->mark_processed(remaining_id));
    auto stored_remaining = OutboxTableTest::_table->find_by_event_id(remaining_id);
    ASSERT_TRUE(stored_remaining);
    EXPECT_EQ(stored_remaining->status(), suye::OutboxStatus::Completed);

    EXPECT_TRUE(OutboxTableTest::_table->claim_batch(1).empty());
}

TEST_F(OutboxTableTest, MarkFailedAndReschedule)
{
    const std::string event_id = suye::uuid();
    auto event = std::make_shared<suye::OutboxEvent>(
        event_id,
        "user",
        "uid-fail",
        "profile_updated",
        "{\"field\":\"nickname\",\"new\":\"fail\"}",
        suye::OutboxStatus::Pending,
        2);

    {
        odb::transaction trans(OutboxTableTest::_db->begin());
        ASSERT_TRUE(OutboxTableTest::_table->record_event(event));
        trans.commit();
    }

    auto batch = OutboxTableTest::_table->claim_batch(1);
    ASSERT_EQ(batch.size(), 1u);
    EXPECT_EQ(batch.front()->event_id(), event_id);

    EXPECT_TRUE(OutboxTableTest::_table->mark_failed(event_id, "first failure", boost::posix_time::seconds(0)));

    auto retry_batch = OutboxTableTest::_table->claim_batch(1);
    ASSERT_EQ(retry_batch.size(), 1u);
    EXPECT_EQ(retry_batch.front()->event_id(), event_id);

    EXPECT_TRUE(OutboxTableTest::_table->mark_failed(event_id, "second failure", boost::posix_time::seconds(0)));

    auto stored = OutboxTableTest::_table->find_by_event_id(event_id);
    ASSERT_TRUE(stored);
    EXPECT_EQ(stored->status(), suye::OutboxStatus::Failed);
    EXPECT_EQ(stored->retry_count(), 2u);

    EXPECT_TRUE(OutboxTableTest::_table->claim_batch(1).empty());
}

TEST_F(OutboxTableTest, RequeueStuckDispatching)
{
    const std::string event_id = suye::uuid();
    auto event = std::make_shared<suye::OutboxEvent>(
        event_id,
        "user",
        "uid-stuck",
        "profile_updated",
        "{\"field\":\"nickname\",\"new\":\"stuck\"}");

    {
        odb::transaction trans(OutboxTableTest::_db->begin());
        ASSERT_TRUE(OutboxTableTest::_table->record_event(event));
        trans.commit();
    }

    auto batch = OutboxTableTest::_table->claim_batch(1);
    ASSERT_EQ(batch.size(), 1u);
    EXPECT_EQ(batch.front()->event_id(), event_id);

    auto requeued = OutboxTableTest::_table->requeue_stuck_events(boost::posix_time::seconds(0));
    EXPECT_EQ(requeued, 1u);

    auto stored = OutboxTableTest::_table->find_by_event_id(event_id);
    ASSERT_TRUE(stored);
    EXPECT_EQ(stored->status(), suye::OutboxStatus::Pending);

    auto next_batch = OutboxTableTest::_table->claim_batch(1);
    ASSERT_EQ(next_batch.size(), 1u);
    EXPECT_EQ(next_batch.front()->event_id(), event_id);
}

int main(int argc, char* argv[])
{
    google::ParseCommandLineFlags(&argc, &argv, true);
    suye::Init_logger(false, "", 0);
    ::testing::InitGoogleTest(&argc, argv);
    return RUN_ALL_TESTS();
}
